package org.apache.beam.examples.PTransforms;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import org.apache.beam.examples.Models.QualityRuleColData;
import org.apache.beam.examples.Models.QualityRuleResultModel;
import org.apache.beam.examples.Models.enums.QualityType;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Distribution;
import org.apache.beam.sdk.metrics.Metrics;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.SimpleFunction;

import java.util.ArrayList;
import java.util.List;

public class QualityRuleDoFn extends DoFn<ArrayList<QualityRuleColData>, QualityRuleResultModel> {

    static final Log log = LogFactory.get();

//    @Override
//    public ArrayList<QualityRuleResultModel> apply(ArrayList<QualityRuleColData> elements) {
//        ArrayList<QualityRuleResultModel> models = new ArrayList<>();
//
//        int count = elements.size();
//        for (QualityRuleColData qualityRuleColData : elements) {
//            count += qualityRuleColData.getDetailModel().getQualityTypes().size();
//        }
//        lineLenDist.update(count);
//
//        for (QualityRuleColData element : elements) {
//            // 取出所有的需要检查的规则
//            List<QualityType> words = element.getDetailModel().getQualityTypes();
//
//            // 循环处理内容生成规则
//            for (QualityType word : words) {
//                if (word != null) {
//                    models.add(word.execute(element));
//                }
//            }
//        }
//        return models;
//    }

    @ProcessElement
    public void processElement(@Element ArrayList<QualityRuleColData> elements, OutputReceiver<QualityRuleResultModel> receiver) {

        log.debug("我感觉你没进来啊！就是没进来");
        int count = elements.size();
        for (QualityRuleColData qualityRuleColData : elements) {
            count += qualityRuleColData.getDetailModel().getQualityTypes().size();
        }
        for (QualityRuleColData element : elements) {
            // 取出所有的需要检查的规则
            List<QualityType> words = element.getDetailModel().getQualityTypes();
            // 循环处理内容生成规则
            for (QualityType word : words) {
                if (word != null) {
                    receiver.output(word.execute(element));
                }
            }
        }

    }
}
